package com.atguigu.flink.datastramapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Created by Smexy on 2023/2/25
 *
 *  把s1传感器，放入主流
 *  把s2传感器，放入支流
 *  把s3传感器，放入支流
 *
 */
public class Demo9_ProcessSideOutPut
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
                常用的类型: Types
                自定义的类型: TypeInfomation.of(Class T)
         */
        OutputTag<WaterSensor> s2 = new OutputTag<>("s2", TypeInformation.of(WaterSensor.class));
        OutputTag<WaterSensor> s3 = new OutputTag<>("s3",TypeInformation.of(WaterSensor.class));

        //主流
        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .process(new ProcessFunction<WaterSensor, WaterSensor>()
            {
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                    if ("s1".equals(value.getId())) {
                        //使用主流输出
                        out.collect(value);
                    } else if ("s2".equals(value.getId())) {
                               /*
                                使用侧(支)流输出
                                output(OutputTag<X> outputTag, X value)
                                    OutputTag: 声明流中输出的数据的类型
                                */
                        ctx.output(s2, value);
                    } else if ("s3".equals(value.getId())) {
                        ctx.output(s3, value);
                    }
                }
            });

        //使用主流获取侧流
        DataStream<WaterSensor> s2DS = ds.getSideOutput(s2);
        DataStream<WaterSensor> s3DS = ds.getSideOutput(s3);

        //主流输出
        ds.print("主流");
        s2DS.keyBy(x->"x").sum("vc").printToErr("侧流s2累加vc");
        s3DS.keyBy(x->"x").max("vc").printToErr("侧流s3求最大值");

        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
